package com.dy.realtime.dwd;

import com.dy.realtime.common.base.BaseSqlApp;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.realtime.dwd
 * @className: DwdTradePrescription
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/11 19:12
 * @version: 1.0
 * todo 交易域处方开单事务事实表
 */

public class DwdTradePrescription extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradePrescription().start(Constant.TOPIC_DWD_TRADE_PRESCRIPTION, Constant.TOPIC_DWD_TRADE_PRESCRIPTION, 4, 10013);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        readOdsDb(tableEnv, Constant.TOPIC_DWD_TRADE_PRESCRIPTION);
        Table prescription = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] as id,\n" +
                "    `data`['create_time'] as create_time,\n" +
                "    `data`['instruction'] as prescription_instruction,\n" +
                "    `data`['status'] as status,\n" +
                "    `data`['total_amount'] as total_amount,\n" +
                "    `data`['consultation_id'] as consultation_id,\n" +
                "    `data`['doctor_id'] as doctor_id,\n" +
                "    `data`['patient_id'] as patient_id\n" +
                "from  topic_db_medical\n" +
                "where `table` = 'prescription'\n" +
                "  and `type` = 'insert';");
//        prescription.execute().print();
        tableEnv.createTemporaryView("prescription", prescription);


        Table prescription_detail = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] as id,\n" +
                "    `data`['create_time'] as create_time,\n" +
                "    `data`['count'] as `count`,\n" +
                "    `data`['instruction'] as prescription_detail_instruction,\n" +
                "    `data`['medicine_id'] as medicine_id,\n" +
                "    `data`['prescription_id'] as prescription_id\n" +
                "from  topic_db_medical\n" +
                "where `table` = 'prescription_detail'\n" +
                "  and `type` = 'insert'");
//        prescription_detail.execute().print();
        tableEnv.createTemporaryView("prescription_detail", prescription_detail);

        Table result = tableEnv.sqlQuery("select\n" +
                "    p.id,\n" +
                "    p.create_time,\n" +
                "    prescription_instruction,\n" +
                "    status,\n" +
                "    total_amount,\n" +
                "    consultation_id,\n" +
                "    `count`,\n" +
                "    prescription_detail_instruction,\n" +
                "    doctor_id,\n" +
                "    patient_id,\n" +
                "    prescription_id    ,\n" +
                "    medicine_id\n" +
                "from  prescription p \n" +
                "       join  prescription_detail pd on p.id = pd.prescription_id");
//        result.execute().print();
        tableEnv.executeSql("create table " + Constant.TOPIC_DWD_TRADE_PRESCRIPTION + "\n" +
                "(\n" +
                "    id                              string,\n" +
                "    create_time                     string,\n" +
                "    prescription_instruction        string,\n" +
                "    status                          string,\n" +
                "    total_amount                    string,\n" +
                "    consultation_id                 string,\n" +
                "    `count`                         string,\n" +
                "    prescription_detail_instruction string,\n" +
                "    doctor_id                       string,\n" +
                "    patient_id                      string,\n" +
                "    prescription_id                      string,\n" +
                "    medicine_id                     string\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_PRESCRIPTION));
        result.executeInsert(Constant.TOPIC_DWD_TRADE_PRESCRIPTION);

    }
}
